package com.liuyu.common.framework.common.event.servic.impl;

import com.google.common.collect.Maps;
import com.liuyu.common.framework.common.event.service.facade.EventPublisherService;
import com.liuyu.common.framework.rocketmq.properties.RocketMqTopic;
import com.liuyu.common.framework.rocketmq.service.facade.MessageProducerService;
import com.liuyu.common.framework.rocketmq.vo.MqMessage;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Map;

/**
 * rocket 事件发布器实现类
 *
 * @author: liuyu
 * @date: 2022/6/8 13:34
 * @version: 1
 */
@Service
public class RocketEventPublisherServiceImpl implements EventPublisherService {
    @Autowired
    private RocketMqTopic rocketMqTopic;

    @Autowired
    private MessageProducerService messageProducerService;

    /**
     * 发送普通的消息
     *
     * @param topic
     * @param message
     * @return void
     * @throws Exception
     * @author liuyu
     * @date 2022/4/24 9:42
     */
    @Override
    public <T> void sendCommonMessage(String topic, MqMessage<T> message) throws Exception {
        messageProducerService.sendMessage(getTopicName(topic), message);
    }

    /**
     * 发送普通的消息
     *
     * @param topic
     * @param message
     * @return void
     * @throws Exception
     * @author liuyu
     * @date 2022/4/24 9:42
     */
    @Override
    public <T> void sendMessage(String topic, MqMessage<T> message) throws Exception {
        messageProducerService.sendMessage(getDirectTopicName(topic), message);
    }

    private String getTopicName(String topic) {
        StringBuffer topicName = new StringBuffer();
        if (rocketMqTopic.isSingleTopic()) {
            if (rocketMqTopic.getTopicMap().containsKey(topic)) {
                topicName.append(rocketMqTopic.getTopicMap().get(topic));
            } else {
                topicName.append(rocketMqTopic.getOrderlyTopicMap().get(topic));
            }
        } else {
            topicName.append(rocketMqTopic.getTopicMap().get(RocketMqTopic.TOPIC_COMMON));
        }
        return topicName.toString();
    }

    private String getDirectTopicName(String topic) {
        if (rocketMqTopic.getTopicMap().containsKey(topic)) {
            return rocketMqTopic.getTopicMap().get(topic);
        } else {
            return rocketMqTopic.getOrderlyTopicMap().get(topic);
        }
    }

    /**
     * 发送普通的消息
     * 通过key 的hash选择一个队列，保证一组数据的局部顺序性。
     *
     * @param message
     * @return void
     * @throws Exception
     * @author liuyu
     * @date 2021/7/29 15:04PayServiceImpl.java
     */
    @Override
    public <T> void sendOrderMessage(String topic, MqMessage<T> message) throws Exception {
        messageProducerService.sendOrderMessage(getDirectTopicName(topic), message);
    }

    /**
     * 发送普通的消息
     * 通过key 的hash选择一个队列，保证一组数据的局部顺序性。
     *
     * @param topic
     * @param message
     * @return void
     * @throws Exception
     * @author liuyu
     * @date 2021/7/29 15:04
     */
    @Override
    public <T> void sendCommonOrderMessage(String topic, MqMessage<T> message) throws Exception {
        messageProducerService.sendOrderMessage(getTopicName(topic), message);
    }

    /**
     * 发送事务消息
     *
     * @param topicName
     * @param message
     * @param object
     * @return org.apache.rocketmq.client.producer.TransactionSendResult
     * @author liuyu
     * @date 2021/12/23 11:05
     */
    @Override
    public <T> TransactionSendResult sendMessageInTransaction(String topicName, MqMessage<T> message, Object object) {
        Map<String, Object> header = createHeader(message);
        return messageProducerService.sendMessageInTransaction(getDirectTopicName(topicName), message, object, header);
    }

    private <T> Map<String, Object> createHeader(MqMessage<T> message) {
        Map<String, Object> header = Maps.newHashMap();
        if (StringUtils.isNotBlank(message.getKey())) {
            header.put(RocketMQHeaders.KEYS, message.getKey());
        }
        if (StringUtils.isNotBlank(message.getTags())) {
            header.put(RocketMQHeaders.TAGS, message.getTags());
        }
        return header;
    }

    /**
     * 发送事务消息
     *
     * @param topicName
     * @param message
     * @param object
     * @return org.apache.rocketmq.client.producer.TransactionSendResult
     * @author liuyu
     * @date 2021/12/23 11:05
     */
    @Override
    public <T> TransactionSendResult sendMessageInTransaction(String topicName, MqMessage<T> message, Object object, Map<String, Object> header) {
        Map<String, Object> headers = createHeader(message);
        headers.putAll(header);
        return messageProducerService.sendMessageInTransaction(getDirectTopicName(topicName), message, object, headers);
    }

    /**
     * 发送事务消息
     *
     * @param topicName
     * @param message
     * @param object
     * @param checkTimeSeconds 回查间隔时间
     * @return org.apache.rocketmq.client.producer.TransactionSendResult
     * @author liuyu
     * @date 2021/12/23 11:05
     */
    @Override
    public <T> TransactionSendResult sendMessageInTransaction(String topicName, MqMessage<T> message, Object object, Integer checkTimeSeconds) {
        Map<String, Object> headers = createHeader(message);
        headers.put(MessageConst.PROPERTY_CHECK_IMMUNITY_TIME_IN_SECONDS, checkTimeSeconds + "");
        return messageProducerService.sendMessageInTransaction(getDirectTopicName(topicName), message, object, headers);
    }


}
